#include "qtc_alarm_kafka/AlarmKafkaConsumer.h"
#include <QDebug>
#include <QJsonDocument>
#include <QJsonObject>
#include <chrono>

// librdkafka C API (Homebrew installs headers under librdkafka/)
#include <librdkafka/rdkafka.h>

namespace qtc_alarm_kafka {

AlarmKafkaConsumer::AlarmKafkaConsumer(const Config& cfg, EventCallback cb)
  : m_cfg(cfg), m_cb(std::move(cb)) {}

AlarmKafkaConsumer::~AlarmKafkaConsumer() { stop(); wait(); }

void AlarmKafkaConsumer::stop() { m_running = false; }

static void set_conf(rd_kafka_conf_t* conf, const char* name, const char* value) {
  char errstr[512];
  if (rd_kafka_conf_set(conf, name, value, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
    qWarning() << "librdkafka conf set failed" << name << value << ":" << errstr;
  }
}

void AlarmKafkaConsumer::run() {
  qInfo() << "AlarmKafkaConsumer starting" << m_cfg.broker << "topic" << m_cfg.topic_config;

  rd_kafka_conf_t* conf = rd_kafka_conf_new();
  set_conf(conf, "bootstrap.servers", m_cfg.broker.toUtf8().constData());
  set_conf(conf, "group.id", m_cfg.group_id.toUtf8().constData());
  set_conf(conf, "auto.offset.reset", "earliest");
  set_conf(conf, "enable.auto.commit", "true");

  rd_kafka_t* rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, nullptr, 0);
  if (!rk) {
    qWarning() << "Failed to create librdkafka consumer";
    return;
  }
  rd_kafka_poll_set_consumer(rk);

  rd_kafka_topic_partition_list_t* topics = rd_kafka_topic_partition_list_new(1);
  rd_kafka_topic_partition_list_add(topics, m_cfg.topic_config.toUtf8().constData(), -1);
  if (rd_kafka_subscribe(rk, topics) != RD_KAFKA_RESP_ERR_NO_ERROR) {
    qWarning() << "Subscribe failed" << m_cfg.topic_config;
    rd_kafka_topic_partition_list_destroy(topics);
    rd_kafka_destroy(rk);
    return;
  }
  rd_kafka_topic_partition_list_destroy(topics);

  while (static_cast<bool>(m_running)) {
    rd_kafka_message_t* msg = rd_kafka_consumer_poll(rk, 200);
    if (!msg) continue;
    if (msg->err) {
      if (msg->err != RD_KAFKA_RESP_ERR__PARTITION_EOF && msg->err != RD_KAFKA_RESP_ERR__TIMED_OUT) {
        qWarning() << "Kafka error:" << rd_kafka_message_errstr(msg);
      }
      rd_kafka_message_destroy(msg);
      continue;
    }

    const char* key_ptr = static_cast<const char*>(msg->key);
    const size_t key_len = msg->key_len;
    const char* val_ptr = static_cast<const char*>(msg->payload);
    const size_t val_len = msg->len;

    QString key = QString::fromUtf8(key_ptr, static_cast<int>(key_len));
    QByteArray payload(val_ptr, static_cast<int>(val_len));

    // Key format: "state:/Accelerator/Area/Device/PV" or "config:/Accelerator/..."
    const int colon = key.indexOf(":");
    if (colon <= 0) { rd_kafka_message_destroy(msg); continue; }
    const QString prefix = key.left(colon);
    const QString path = key.mid(colon+1); // starts with /Accelerator/...

    // Extract PV name after the root config
    QString pv = path;
    // Remove leading "/" and root filter if present
    if (pv.startsWith('/')) pv = pv.mid(1);
    if (!m_cfg.root_filter.isEmpty() && pv.startsWith(m_cfg.root_filter + '/')) {
      pv = pv.mid(m_cfg.root_filter.size() + 1);
    }

    // Parse JSON value
    QJsonDocument doc = QJsonDocument::fromJson(payload);
    QJsonObject obj = doc.isObject() ? doc.object() : QJsonObject{};

    qtc_alarm_core::AlarmEvent ev;
    ev.type = prefix;
    ev.config = m_cfg.root_filter;
    ev.pv = pv.replace('/', ':');
    ev.root = m_cfg.root_filter;
    ev.severity = obj.value("severity").toString();
    ev.message = obj.value("message").toString();
    ev.state = obj.value("state").toString();
    // Parse time from either numeric 'time_epoch_ms', numeric 'time', or object {seconds,nano}
    ev.time_epoch_ms = obj.value("time_epoch_ms").toVariant().toLongLong();
    if (ev.time_epoch_ms == 0) {
      const auto time_val = obj.value("time");
      if (time_val.isDouble()) {
        ev.time_epoch_ms = time_val.toVariant().toLongLong();
      } else if (time_val.isObject()) {
        const auto tobj = time_val.toObject();
        const qint64 seconds = tobj.value("seconds").toVariant().toLongLong();
        const qint64 nanos = tobj.value("nano").toVariant().toLongLong();
        if (seconds > 0 || nanos > 0) {
          ev.time_epoch_ms = seconds * 1000 + nanos / 1000000;
        }
      }
    }

    if (m_cb) m_cb(ev);

    rd_kafka_message_destroy(msg);
  }

  rd_kafka_unsubscribe(rk);
  rd_kafka_consumer_close(rk);
  rd_kafka_destroy(rk);
  qInfo() << "AlarmKafkaConsumer stopped";
}

} // namespace qtc_alarm_kafka